package com.swjd.transaction;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;

//模拟事务 发送消息
public class Producer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        //消息发送者（生产着)
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //发送给服务器地址
        producer.setNamesrvAddr("localhost:9876");
        //设置事务监听
        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //模拟把消息保存到数据库里
                System.out.println("正常执行事务过程");
                // ROLLBACK_MESSAGE回滚 COMMIT_MESSAGE 提交消息 UNKNOW未响应消息结果，将执行事务补偿
                return LocalTransactionState.UNKNOW;
            }

            //事务补偿过程
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("执行事务补偿过程");
                //模拟数据库操作
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        //启动服务
        producer.start();
        //创建要发送的消息·
        for (int i = 0;i<10; i++){
            String msg = ("这是第"+i+"条消息");
            Message message = new Message("topic12", "tag1", msg.getBytes());
            //发送事务的消息
            TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);
            //输出发送结果
            System.out.println(sendResult);
        }
        //生产者不能关闭，关闭后续事务无法执行
       // producer.shutdown();
    }
}
